package stc.skymobi.netty.handler.codec.tlv;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.AttributeKey;

import java.beans.PropertyEditor;
import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.List;

import javax.management.MBeanInfo;
import javax.management.MBeanOperationInfo;
import javax.management.MBeanParameterInfo;
import javax.management.MBeanServerConnection;
import javax.management.ObjectName;

import org.jocean.util.DefaultPropertyEditorSource;
import org.jocean.util.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import stc.skymobi.bean.bytebean.core.BeanFieldCodec;
import stc.skymobi.bean.tlv.TLVSignal;
import stc.skymobi.bean.tlv.decode.TLVDecoderOfBean;
import stc.skymobi.bean.util.meta.Int2TypeMetainfo;
import stc.skymobi.transport.protocol.xip.XipHeader;
import stc.skymobi.util.ByteUtils;
import stc.skymobi.util.MutableIdentifyable;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

/**
 * 非线程安全
 */
public class NettyMixedTcpDecoder extends ByteToMessageDecoder {

    private static final Logger logger = LoggerFactory.getLogger(NettyMixedTcpDecoder.class);

    private static final AttributeKey<XipHeader> XIP_HEADER_ATTRIBUTE_KEY = AttributeKey.valueOf("XIP_HEADER_ATTRIBUTE_KEY");

    private BeanFieldCodec xipBeanCodec;
    private Int2TypeMetainfo xipTypeMetainfo;
    private TLVDecoderOfBean tlvBeanDecoder;
    private Int2TypeMetainfo tlvTypeMetainfo;

    //大于1M的数据包可能是问题数据包
    private int maxMessageLength = 1024 * 1024;

    private int dumpBytes = 256;
    private boolean isDebugEnabled;

    private static final long SKY_PROTOCAL_ID =
            (0x8AL << 56)
                    | (0xEDL << 48)
                    | (0x9CL << 40)
                    | (0xF3L << 32)
                    | (0x7EL << 24)
                    | (0x32L << 16)
                    | (0xB9L << 8);

    @Override
    public void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
        XipHeader header = ctx.attr(XIP_HEADER_ATTRIBUTE_KEY).get();
        if (null == header) {
            if (buffer.readableBytes() < XipHeader.SSIP_HEADER_LENGTH) {
                return;
            }

            int headerSize = buffer.getLong(0) == SKY_PROTOCAL_ID ? XipHeader.SAIP_HEADER_LENGTH : XipHeader.SSIP_HEADER_LENGTH;
            if (buffer.readableBytes() < headerSize) {
                return;
            }

            if ( logger.isDebugEnabled() && isDebugEnabled ) {
            	logger.debug("parse header... try parse...");
            }

            byte[] headerBytes = new byte[headerSize];
            buffer.readBytes(headerBytes);
            if (logger.isDebugEnabled() && isDebugEnabled) {
                logger.debug("header raw bytes -->");
                logger.debug(ByteUtils.bytesAsHexString(headerBytes, dumpBytes));
            }
            header = (XipHeader) xipBeanCodec.decode(
                    xipBeanCodec.getDecContextFactory().createDecContext(
                            headerBytes, XipHeader.class, null, null)).getValue();
            if (logger.isDebugEnabled() && isDebugEnabled) {
                logger.debug("header-->" + header);
            }

            if (header.getMessageLength() < 0) {
                logger.error("header.messageLength ({}) < 0, so drop this connection {}.\r\ndump bytes received:\r\n{}"
                        , header.getMessageLength(), ctx.channel(), ByteUtils.bytesAsHexString(headerBytes, dumpBytes));
                ctx.channel().close().await();
                throw new RuntimeException("messageLength too large:" + header.getMessageLength());
            }
            if (maxMessageLength > 0) {
                if (header.getMessageLength() > maxMessageLength) {
                    logger.error("header.messageLength ({}) exceed maxMessageLength[{}], so drop this connection{}.\r\ndump bytes received:\r\n{}"
                            , header.getMessageLength(), maxMessageLength, ctx.channel(), ByteUtils.bytesAsHexString(headerBytes, dumpBytes));
                    ctx.channel().close().await();
                    throw new RuntimeException("messageLength too large:" + header.getMessageLength());
                }
            }

            ctx.attr(XIP_HEADER_ATTRIBUTE_KEY).set(header);
        }

        if (buffer.readableBytes() < header.getMessageLength()) {
            logger.debug("readableBytes {} smaller than packageLength {},waiting for remain bytes", buffer.readableBytes(), header.getMessageLength());
            return;
        }

        //为下一次在同一ctx上进行xip接受初始化环境
        ctx.attr(XIP_HEADER_ATTRIBUTE_KEY).remove();

        final byte[] bytes = new byte[header.getMessageLength()];
        buffer.readBytes(bytes);

        if (logger.isTraceEnabled() && isDebugEnabled) {
            logger.trace("body raw bytes \r\n{}",
                    ByteUtils.bytesAsHexString(bytes, dumpBytes));
        }

        if ( 1 == header.getBasicVer() || 2 == header.getBasicVer() ) {
        	final Object outObj = decodeXipOrTlvMessage(header, bytes);
        	if ( null != outObj ) {
        		out.add(outObj);
        	}
        }
        else if ( 100 == header.getBasicVer() ) {
        	final Object outObj = processInnerMsg( header, bytes, ctx );
        	if ( null != outObj ) {
        		out.add(outObj);
        	}
        }
        else  {
            logger.error("invalid basic ver, while header is {}", header);
            logger.error("raw body bytes is {}",
                    ByteUtils.bytesAsHexString(bytes, header.getMessageLength()));
            throw new RuntimeException("invalid basic ver {" + header.getBasicVer() + "}");
        }
    }

	private Object processInnerMsg(final XipHeader header, final byte[] bytes, final ChannelHandlerContext ctx) {
		try {
			final Object obj = JSON.parse(new String(bytes, "UTF-8"));
			if ( obj instanceof JSONObject) {
				final JSONObject json = (JSONObject)obj;
				final String msgType = json.getString("_msgtype");
				if ( "feedbackResult".equals(msgType) ) {
					return json;
				}
				else if ( "invokeMBean".equals(msgType) ) {
					final JSONObject response = doInvokeMBean(json);
					response.put("_transactionId", json.getString("_transactionId"));
					//	设置msgtype 为 feedbackResult
					response.put("_msgtype", "feedbackResult");
					ctx.channel().writeAndFlush(response);
				}
				else {
					//	error
					final JSONObject response = genJsonResponse(false, null, "unknown _msgtype: " + msgType);
					response.put("_transactionId", json.getString("_transactionId"));
					response.put("_msgtype", "feedbackResult");
					ctx.channel().writeAndFlush(response);
				}
			}
			else {
				logger.warn("not except JSON Object {}, just ignore", obj);
			}
		}
		catch (Exception e) {
			logger.warn("exception when processInnerMsg, detail: {}", 
					ExceptionUtils.exception2detail(e));
		}
		
		return	null;
	}

	private static final DefaultPropertyEditorSource _EDITSRC = new DefaultPropertyEditorSource();
	
	private JSONObject doInvokeMBean(final JSONObject json) {
		try {
			final ObjectName objName = ObjectName.getInstance(json
					.getString("objectName"));
			final String operationName = json.getString("operationName");
			final JSONArray paramsArray = json.getObject("params", JSONArray.class);
			final int paramsCount = (null != paramsArray) ? paramsArray.size() : 0;

			final MBeanServerConnection mbsc = ManagementFactory
					.getPlatformMBeanServer();
			final MBeanOperationInfo opinfo = getOperationInfo(objName,
					operationName, paramsCount, mbsc);
			if (null != opinfo) {
				final Object retval = mbsc.invoke(objName, operationName,
						genParamObjects(paramsArray, opinfo.getSignature()),
						genSignatures(opinfo.getSignature()));
				return genJsonResponse(true, retval, null);
			}
			else {
				return genJsonResponse(false, null, objName+"."+operationName + " not found.");
			}
		} catch (Exception e) {
			return	genJsonResponse(false, null, ExceptionUtils.exception2detail(e));
		}
	}

	private JSONObject genJsonResponse(final boolean isSucceed, final Object retval, final String reason) {
		return new JSONObject() {
			private static final long serialVersionUID = 1L;
		{
			this.put("isSucceed", isSucceed);
			if ( null != retval ) {
				this.put("retval", JSON.toJSONString(retval));
			}
			if ( null != reason ) {
				this.put("reason", reason);
			}
		}};
	}

	private String[] genSignatures(final MBeanParameterInfo[] signatures) {
		if ( (null == signatures) ||  (0 == signatures.length)) {
			return	null;
		}
		return new ArrayList<String>() {
			private static final long serialVersionUID = 1L;
		{
			for ( MBeanParameterInfo paraminfo :signatures ) {
				this.add( paraminfo.getType());
			}
		}}.toArray(new String[0]);
	}

	private Object[] genParamObjects(
			final JSONArray params,
			final MBeanParameterInfo[] signatures) {
		if ( (null == params) || ( 0 == params.size())) {
			return	null;
		}
		
		return new ArrayList<Object>() {
			private static final long serialVersionUID = 1L;
		{
			for ( int idx = 0; idx < params.size(); idx++) {
				final PropertyEditor editor = _EDITSRC.getPropertyEditor(signatures[idx].getType());
				editor.setAsText(params.getString(idx));
				this.add(editor.getValue());
			}
		}}.toArray();

	}

	/**
	 * @param objname
	 * @param operation
	 * @param paramsCount
	 * @param mbsc
	 * @throws Exception
	 */
	private MBeanOperationInfo getOperationInfo(
			final ObjectName objname,
			final String operation, 
			final int paramsCount,
			final MBeanServerConnection mbsc) throws Exception {
		final MBeanInfo info = mbsc.getMBeanInfo(objname);
		final MBeanOperationInfo[] ops = info.getOperations();
		for (int idx = 0; idx < ops.length; idx++) {
			if (ops[idx].getName().equals(operation)
					&& (ops[idx].getSignature().length == paramsCount)) {
				return ops[idx];
			}
		}
		return null;
	}

	/**
	 * @param header
	 * @param bytes
	 */
	private Object decodeXipOrTlvMessage(final XipHeader header, final byte[] bytes) {
		MutableIdentifyable identifyable = null;
        switch( header.getBasicVer()) {
        case	1: {
            final Class<?> type = xipTypeMetainfo.find(header.getMessageCode());
            if (null == type) {
                // throw new RuntimeException("unknow message code:" + header.getMessageCode());
            	logger.error("unknown xip's message code: {}, just ignore this message.",  header.getMessageCode());
            	return	null;
            }
            identifyable = (MutableIdentifyable) xipBeanCodec.decode(
                    xipBeanCodec.getDecContextFactory().createDecContext(
                            bytes, type, null, null)).getValue();
        	}
            break;
        case	2: {
            final Class<?> type = tlvTypeMetainfo.find(header.getMessageCode());
            if (null == type) {
                //throw new RuntimeException("unknow tag:" + header.getMessageCode());
            	logger.error("unknown tlv's message code: {}, just ignore this message.",  header.getMessageCode());
            	return	null;
            }
            identifyable = (MutableIdentifyable) tlvBeanDecoder.decode(
                    bytes.length, bytes,
                    tlvBeanDecoder.getDecodeContextFactory().createDecodeContext(type, null));
            if (identifyable instanceof TLVSignal) {
                ((TLVSignal) identifyable).setSourceId(header.getSourceId());
            }
        	}
        	break;
        default:
            throw new RuntimeException("invalid basic ver {" + header.getBasicVer() + "}");
        }

        identifyable.setIdentification(header.getTransactionAsUUID());

        if (logger.isDebugEnabled() && isDebugEnabled) {
            logger.debug("signal-->" + identifyable);
        }
        
        return	identifyable;
	}

    /**
     * @param dumpBytes the dumpBytes to set
     */
    public void setDumpBytes(int dumpBytes) {
        this.dumpBytes = dumpBytes;
    }

    public int getDumpBytes() {
        return dumpBytes;
    }

    public boolean isDebugEnabled() {
        return isDebugEnabled;
    }

    public void setDebugEnabled(boolean isDebugEnabled) {
        this.isDebugEnabled = isDebugEnabled;
    }

    /**
     * @return the esbBeanCodec
     */
    public BeanFieldCodec getXipBeanCodec() {
        return xipBeanCodec;
    }

    /**
     * @param xipBeanCodec the esbBeanCodec to set
     */
    public void setXipBeanCodec(BeanFieldCodec xipBeanCodec) {
        this.xipBeanCodec = xipBeanCodec;
    }

    /**
     * @return the esbTypeMetainfo
     */
    public Int2TypeMetainfo getXipTypeMetainfo() {
        return xipTypeMetainfo;
    }

    /**
     * @param xipTypeMetainfo the esbTypeMetainfo to set
     */
    public void setXipTypeMetainfo(Int2TypeMetainfo xipTypeMetainfo) {
        this.xipTypeMetainfo = xipTypeMetainfo;
    }

    /**
     * @return the maxMessageLength
     */
    public int getMaxMessageLength() {
        return maxMessageLength;
    }

    /**
     * @param maxMessageLength the maxMessageLength to set
     */
    public void setMaxMessageLength(int maxMessageLength) {
        this.maxMessageLength = maxMessageLength;
    }

    public void setTlvBeanDecoder(TLVDecoderOfBean tlvBeanDecoder) {
        this.tlvBeanDecoder = tlvBeanDecoder;
    }

    public void setTlvTypeMetainfo(Int2TypeMetainfo tlvTypeMetainfo) {
        this.tlvTypeMetainfo = tlvTypeMetainfo;
    }
}
